/**
 * Created by lzfto
 * Created time: 2025/05/25
 */
package org.lzfto.flink.demo.business.flinkTest.test;

import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.lzfto.flink.demo.business.flinkTest.test.entity.TestInfo;
import org.lzfto.flink.demo.common.result.PubResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Flink测试服务
 */
public class SimpleTestService implements AutoCloseable {

	private static Logger LOGGER = LoggerFactory.getLogger(SimpleTestService.class);
	/**
	 * 以服务实现编号为key，value是FlinkTestInterface的实现bean
	 */
	private Map<Integer, FlinkTestInterface> tests;

	private List<TestInfo> testInfoLst;


	public SimpleTestService(List<FlinkTestInterface> testList) {
		LOGGER.info("初始化Flink测试任务");
		testList.sort((t1,t2)->{
			return t1.getTestInfo().no().compareTo(t2.getTestInfo().no());
		});
		// 使用反射查找所有实现了FlinkTestInterface的类，并创建实例
		this.testInfoLst = new LinkedList<>();
		this.tests = new HashMap<>();		
		for (FlinkTestInterface test : testList) {
			TestInfo info = test.getTestInfo();
			testInfoLst.add(info);
			tests.put(test.getTestInfo().no(), test);
			LOGGER.info("打印Flink测试任务基本信息:no={},name={},comments={}", info.no(), info.name(), info.comments());
		}

	}

	public List<TestInfo> getTasks() {
		return this.testInfoLst;
	}
	
	public TestInfo getTestInfo(Integer testNo) {
		FlinkTestInterface test=this.tests.get(testNo);
		if (test==null) {
			return null;
		}
		else {
			return test.getTestInfo();	
		}
	}

	/**
	 * 运行Flink测试
	 * @param testNo 测试序号
	 * @return
	 */
	public Object runTest(int testNo, String jobName) {
		if (testNo == -1) {
			return "没有做选择！";
		}
		try {
			FlinkTestInterface test = this.tests.get(testNo);
			if (test == null) {
				return "不存在编号为" + testNo + "的测试任务";
			}
			Object result=test.run(jobName);
			TestInfo testInfo=test.getTestInfo();
			LOGGER.info("测试作业(no={},name={},说明={})已经提交运行",testInfo.no(),testInfo.name(),testInfo.comments());
			return result;
		} catch (Exception e) {
			LOGGER.error("执行出错:{}", e);
			return e.getMessage();
		}
	}

	/**
	 * 专门用于停止本地的集群，生产环境不要这么用
	 */
	public PubResult<Boolean> cleanTest(int taskId) {
		FlinkTestInterface test = this.tests.get(taskId);
		if (test == null) {
			return PubResult.error("不存在实例");
		}

		TestInfo info = test.getTestInfo();
		try {
			return test.clean();
		} catch (Exception e) {
			LOGGER.error("停止测试(id={},name={})失败:{}", info.no(), info.name(), e.getMessage());
			return PubResult.error("停止作业(id=" + info.no() + ",name=" + info.name() + ")失败:" + e.getMessage());
		}
	}

	@Override
	public void close() throws Exception {
		Iterator<Entry<Integer, FlinkTestInterface>> it = this.tests.entrySet().iterator();
		while (it.hasNext()) {
			FlinkTestInterface test = it.next().getValue();
			TestInfo info = test.getTestInfo();
			try {
				if (test != null) {
					test.clean();
				}
			} catch (Exception e) {
				LOGGER.error("停止测试(id={},name={})失败:{}", info.no(), info.name(), e.getMessage());
			}
		}
	}
}
